package com.xqcx.libra.util;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.google.common.collect.Maps;
import com.xqcx.libra.config.MqConfig;
import com.xqcx.libra.event.MessageEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;


@Slf4j
@Component
public class StorageMessage {

    @Autowired
    private MqConfig mqConfig;

    public Map<String, BlockingQueue<Map<String, Object>>> tm = Maps.newHashMap();

    public void distributeMessage(Message message) {
        List<CanalEntry.Entry> entrys = message.getEntries();
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }
            CanalEntry.Header header = entry.getHeader();
            String tableName = header.getTableName();
            CanalEntry.RowChange rowChange = null;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
            }
            CanalEntry.EventType eventType = rowChange.getEventType();
            if (eventType == CanalEntry.EventType.QUERY || rowChange.getIsDdl()) {
                continue;
            }
            Map<String, Object> map = Maps.newHashMap();
            map.put("logfileName", header.getLogfileName());
            map.put("logfileOffset", header.getLogfileOffset());
            map.put("executeTime", header.getExecuteTime());
            map.put("schemaName", header.getSchemaName());
            map.put("tableName", header.getTableName());
            map.put("eventLength", header.getEventLength());
            map.put("eventType", header.getEventType());
            map.put("endTime", System.currentTimeMillis());
            List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
            for (CanalEntry.RowData rowData : rowDatasList) {
                Map<String, Object> date = Maps.newHashMap();
                date.putAll(map);
                List<CanalEntry.Column> columns = null;
                if (eventType == CanalEntry.EventType.DELETE) {
                    date.put("eventType", "delete");
                    columns = rowData.getBeforeColumnsList();
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    date.put("eventType", "insert");
                    columns = rowData.getAfterColumnsList();
                } else {
                    date.put("eventType", "update");
                    columns = rowData.getAfterColumnsList();
                }
                for (CanalEntry.Column column : columns) {
                    // 主键Id
                    if (column.getIsKey()) {
                        date.put("primaryKey", column.getValue());
                    }
                    String columnName = processField(column.getName());
                    date.put(columnName, column.getValue());
                    date.put(columnName + "Updated", column.getUpdated());
                }
                try {
                    String topicName = getTopicName(tableName);
                    BlockingQueue<Map<String, Object>> queues = tm.get(topicName);
                    if (queues == null) {
                        Object lock = new Object();
                        synchronized (lock) {
                            if (tm.get(topicName) == null) {
                                // 创建表对应的mq信息和队列
                                queues = new LinkedBlockingQueue<Map<String, Object>>(mqConfig.getTaskNum());
                                tm.put(topicName, queues);
                                createEvent(topicName, queues);
                            } else {
                                queues = tm.get(topicName);
                            }
                        }
                    }
                    queues.put(date);
                } catch (Exception e) {
                    log.error("StorageMessage.distributeMessage Exception：", e);
                }
            }
        }
    }

    /**
     * 新起线程消费
     *
     * @param: [topicName, queues]
     * @return: void
     * @author: xuziwen
     * @Date: 2017-06-30 17:19
     */
    private synchronized void createEvent(String topicName, BlockingQueue<Map<String, Object>> queues) throws Exception {
        // 新起线程消费
        String topic = mqConfig.getTopicPrefix() + topicName;
        String tag = mqConfig.getTagPrefix() + topicName;
        String producerGroup = topicName + mqConfig.getSuffixProducer();
        // 创建producer
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(mqConfig.getNameSrvAddress());
        producer.setPollNameServerInteval(10000);
        producer.start();
        for (int i = 0; i < mqConfig.getTaskNum(); i++) {
            MessageEvent event = new MessageEvent(topic, tag, queues, producer);
            event.start();
        }
    }

    /**
     * 表字段转驼峰
     *
     * @return:
     * @author: xuziwen
     * @Date: 14:49 2017/9/26
     */
    public static String processField(String field) {
        StringBuffer sb = new StringBuffer(field.length());
        String[] fields = field.split("_");
        String temp = null;
        sb.append(fields[0]);
        for (int i = 1; i < fields.length; i++) {
            temp = fields[i].trim();
            sb.append(temp.substring(0, 1).toUpperCase()).append(
                    temp.substring(1));
        }
        return sb.toString();
    }

    /**
     * 拆分表发送相同topic
     *
     * @return:
     * @author: xuziwen
     * @Date: 14:48 2017/9/26
     */
    private String getTopicName(String tableName) {
        String topicName = ShardTableNameProcess.converter(tableName);
        for (Map.Entry<String, String> entry : mqConfig.getTableTopic().entrySet()) {
            String key = entry.getKey();
            String value = entry.getValue();
            if (topicName.equals(key)) {
                topicName = value;
                break;
            }
        }
        return topicName;
    }
}
